package com.sh.rpc.registry;


import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.ConcurrentHashSet;
import cn.hutool.cron.CronUtil;
import cn.hutool.cron.task.Task;
import cn.hutool.json.JSONUtil;
import com.google.common.primitives.Bytes;
import com.sh.rpc.confige.RegistryConfig;
import com.sh.rpc.model.ServiceMetaInfo;
import io.etcd.jetcd.*;
import io.etcd.jetcd.options.GetOption;
import io.etcd.jetcd.options.PutOption;
import io.etcd.jetcd.watch.WatchEvent;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

/**
 * etcd 注册中心
 *
 * @author sh
 */
@Slf4j
public class EtcdRegistry implements Registry {


    private Client client;  // etcd 客户端
    private KV kvClient;    // etcd kv 客户端

    /**
     * 根节点  数据存储端的前缀
     */
    private String ETCD_ROOT_PATH = "/rpc/";

    /**
     * 注册的节点key的集合
     */
    private final Set<String> localRegisterNodeKeySet = new HashSet<>();

    /**
     * 注册中心服务缓存
     */
    private final RegistryServiceCache registryServiceCache = new RegistryServiceCache();

    /**
     * 监听的key集合
     * 消费端调用服务发现的时候就把 key 添加到这个集合中  然后一直监听着
     * 如果服务端把这个服务注销了，及时更新本地的缓存列表
     */
    private final Set<String> watchingKeySet = new ConcurrentHashSet<>();


    @Override
    public void init(RegistryConfig registryConfig) {
        client = Client.builder()
                .endpoints(registryConfig.getAddress())
                .connectTimeout(Duration.ofMillis(registryConfig.getTimeout())).build();
        kvClient = client.getKVClient();
//        heatBeat();  // 启动心跳
    }

    @Override
    public void register(ServiceMetaInfo serviceMetaInfo) throws Exception {
        // 服务注册
        Lease leaseClient = client.getLeaseClient();    // 租约客户端

        long leaseId = leaseClient.grant(1000).get().getID();    // 租约id

        // 设置键值对
        //   /rpc/test:1.0:127.0.0.1:3827
        String registryKey = ETCD_ROOT_PATH + serviceMetaInfo.getServiceNodeKey();

        ByteSequence key = ByteSequence.from(registryKey, StandardCharsets.UTF_8);   // /rpc/test:1.0:127.0.0.1:3827
        //   {"serviceName":"test","serviceVersion":"1.0","serviceHost":"127.0.0.1","servicePort":3827,"serviceGroup":"default"}
        ByteSequence value = ByteSequence.from(JSONUtil.toJsonStr(serviceMetaInfo), StandardCharsets.UTF_8);

        // 与租约时间绑定
        PutOption putOption = PutOption.builder().withLeaseId(leaseId).build();
        kvClient.put(key, value, putOption).get();

        // 添加到本地注册的节点key的集合中   /rpc/test:1.0:127.0.0.1:3827
        localRegisterNodeKeySet.add(registryKey);
    }

    @Override
    public void unregister(ServiceMetaInfo serviceMetaInfo) {
        String registerKey = ETCD_ROOT_PATH + serviceMetaInfo.getServiceNodeKey();
        // 根据 key 删除
        kvClient.delete(ByteSequence.from(ETCD_ROOT_PATH + serviceMetaInfo.getServiceNodeKey(), StandardCharsets.UTF_8));
        // 从本地注册的节点key的集合中删除
        localRegisterNodeKeySet.remove(registerKey);
    }

    @Override
    public List<ServiceMetaInfo> serviceDiscovery(String serviceKey) {
        // 先走本地缓存
        List<ServiceMetaInfo> cachedServiceMetaInfo = registryServiceCache.readCache();
        if (cachedServiceMetaInfo != null) {
            return cachedServiceMetaInfo;
        }

        String searchPrefix = ETCD_ROOT_PATH + serviceKey + "/";   // 因为 serviceKey 没有 / 结尾
        try {
            GetOption getOption = GetOption.builder().isPrefix(true).build();   // 设置前缀查询
            List<KeyValue> keyValues = kvClient.get(
                            ByteSequence.from(searchPrefix, StandardCharsets.UTF_8),
                            getOption)
                    .get()
                    .getKvs();
            List<ServiceMetaInfo> serviceMetaInfoList = keyValues.stream()
                    .map(keyValue -> {
                        String key = keyValue.getKey().toString(StandardCharsets.UTF_8);
                        watch(key);  // 发现时就监听这个节点
                        String value = keyValue.getValue().toString(StandardCharsets.UTF_8);
                        return JSONUtil.toBean(value, ServiceMetaInfo.class);
                    }).collect(Collectors.toList());

            // 写入本地缓存
            registryServiceCache.writeCache(serviceMetaInfoList);
            return serviceMetaInfoList;
        } catch (Exception e) {
            throw new RuntimeException("服务发现失败", e);
        }
    }

    @Override
    public void destroy() {
        log.warn("etcd 服务销毁");
        // 遍历本地注册的节点key的集合，对每个节点进行注销
        for (String key : localRegisterNodeKeySet) {
            try {
                kvClient.delete(ByteSequence.from(key, StandardCharsets.UTF_8)).get();
                log.info("etcd 服务注销成功，key:{}", key);
            } catch (Exception e) {
                log.error("etcd 服务注销失败", e);
                throw new RuntimeException(key + "etcd 服务注销失败", e);
            }
        }


        if (kvClient != null) {
            kvClient.close();
        }
        if (client != null) {
            client.close();
        }
    }

    /**
     * 心跳检测
     */
    @Override
    public void heatBeat() {
        CronUtil.schedule("*/10 * * * * ?", new Task() {
            @Override
            public void execute() {
                // 遍历本地注册的节点key的集合，对每个节点进行心跳检测
                for (String key : localRegisterNodeKeySet) {
                    try {
                        List<KeyValue> keyValues = kvClient.get(ByteSequence.from(key, StandardCharsets.UTF_8)).get().getKvs();
                        if (CollUtil.isEmpty(keyValues)) {  // 如果没有找到对应的key，则过期
                            continue;
                        }
                        KeyValue keyValue = keyValues.get(0);
                        String value = keyValue.getValue().toString(StandardCharsets.UTF_8);  // {xxx:xx, xxx:xx}
                        ServiceMetaInfo serviceMetaInfo = JSONUtil.toBean(value, ServiceMetaInfo.class);
                        log.info("etcd 心跳检测成功，key:{}", key);
                        register(serviceMetaInfo);
                    } catch (Exception e) {
                        log.error("etcd 心跳检测失败", e);
                        throw new RuntimeException(key + "etcd 心跳检测失败", e);
                    }
                }
            }
        });
        // 精确度 秒
        CronUtil.setMatchSecond(true);
        CronUtil.start();
    }

    @Override
    public void watch(String serviceNodeKey) {
        Watch watchClient = client.getWatchClient();
        boolean newWatch = watchingKeySet.add(serviceNodeKey);  // 该key是否在被监听
        if (newWatch) {
            watchClient.watch(ByteSequence.from(serviceNodeKey, StandardCharsets.UTF_8), response -> {
                for (WatchEvent event : response.getEvents()) {
                    switch (event.getEventType()) {
                        case DELETE:
                            // todo 支持多个服务
                            registryServiceCache.clearCache();
                            break;
                        case PUT:
                        default:
                            break;
                    }
                }
            });
        }
    }
}
